package task2

import java.util.concurrent.TimeUnit

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
import org.apache.spark.internal.Logging
import org.apache.spark.util.SerializableConfiguration
import org.apache.spark.{SparkConf, SparkContext}
import task2.util.{FilePathHelper, OptionsParser}

/**
 * Spark 实现 Distcp
 */
object SparkDistcp extends Logging {

  def main(args: Array[String]): Unit = {
    // 解析传入参数
    val option = OptionsParser.parse(args)

    // 构建 Spark 环境，由于环境问题只能跑 local 模式，暂时以 local[xxx] 的思路控制并发度
    val conf = new SparkConf()
      .setMaster(s"local[${option.maxConcurrence}]")
      .setAppName("InvertedIndex")
    val sc = new SparkContext(conf)

    // 创建参数解析类以及 Hadoop Configuration 的广播变量
    val optionBroadcast = sc.broadcast(option)
    val hadoopConf = sc.broadcast(new SerializableConfiguration(sc.hadoopConfiguration))

    // 检查并创建所需文件路径
    FilePathHelper.checkSrcAndCreateTarget(option.source, option.target)
    // 构建所需拷贝文件路径列表
    val copyList = FilePathHelper.buildSrcToTargetList(option.source, option.target)

    // 并发拷贝,此处设置 9 个 partition，在 WebUI 界面验证并发度是否为 3
    val filesRDD = sc.parallelize(copyList, 9)
    filesRDD.foreachPartition(files => {
      val _hadoopConf = hadoopConf.value.value
      // 构建拷贝时所需的参数
      val srcFs = FileSystem.get(_hadoopConf)
      files.foreach(src2target => {
        try {
          val srcPath = new Path(src2target._1)
          val dstPath = new Path(src2target._2)
          val dstFs = dstPath.getFileSystem(_hadoopConf)
          // 开始进行拷贝
          FileUtil.copy(srcFs, srcPath, dstFs, dstPath, false, true, _hadoopConf)
        } catch {
          // 如果 ignoreFailures 为 false，则抛出异常，结束当前 task
          case e: Exception =>
            if (!optionBroadcast.value.ignoreFailures) throw e
            else logError(e.toString)
        }
      })
    })
    // 避免 WebUI 界面关闭
    TimeUnit.DAYS.sleep(1)
  }

}
